package com.cqc.rxjava01.activity;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.view.View;

import com.cqc.rxjava01.R;
import com.cqc.rxjava01.util.LogUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Single;
import io.reactivex.SingleObserver;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

/**
 * 操作符
 */
public class RxJavaOperator02Activity extends AppCompatActivity {

    private Disposable disposable;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_operator02);
    }


    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (disposable != null && disposable.isDisposed()) {
            disposable.dispose();
        }
    }

    /**
     * FlatMap:把上游发送的数据拆分成多个后发送给下游
     *
     * @param view
     */
    public void rxJava01(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer s) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 2; i++) {
                    list.add("5--String--" + i);
                }
                return Observable.fromIterable(list).delay(2000, TimeUnit.SECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtil.d("s=" + s);
            }
        });
    }

    /**
     * 把2个Observable的数据整合一起，发送给观察者
     *
     * @param view
     */
    public void rxJava02(View view) {
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(888);
            }
        });

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("数字:");
            }
        });

        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String s) throws Exception {
                return s + integer;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        LogUtil.d("s=" + s);
                    }
                });
    }

    /**
     * interval:间隔发送
     *
     * @param view
     */
    public void rxJava03(View view) {
        disposable = Flowable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        LogUtil.d("doOnNext--aLong=" + aLong);
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        LogUtil.d("Consumer--aLong=" + aLong);
                    }
                });
    }

    /**
     * distinct
     *
     * @param view
     */
    public void rxJava04(View view) {
        Observable.just(1, 2, 3, 4, 2, 3)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtil.d("Consumer--integer=" + integer);
                    }
                });
    }

    /**
     * filter
     *
     * @param view
     */
    public void rxJava05(View view) {
        Observable.just(5, 25, 30, 18, 15, 20)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer > 15;
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                LogUtil.d("Consumer--integer=" + integer);
            }
        });
    }

    /**
     * buffer:
     * skip:表示步长。第一次index=0,第二次index=0+skip,第三次index=0+skip+skip. index是角标
     * count:表示一次取几个元素
     * 我理解window的功能和buffer类似 不过buffer 把数据源分割成list 而window把数据源分割成了observable
     *
     * @param view
     */
    public void rxJava06(View view) {
        Observable.just(1, 2, 3, 4, 5)
                .buffer(3, 1)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        LogUtil.d("Consumer--integers.size()=" + integers.size() + "\n");
                        LogUtil.d("buffer value:" + "\n");
                        for (Integer integer : integers) {
                            LogUtil.d(integer + "" + "\n");
                        }
                        LogUtil.d("\n");
                    }
                });
    }

    /**
     * doOnNext：上游发送数据后，下游接收数据前。如果我们想做点什么，那么可以在这个方法中操作。
     *
     * @param view
     */
    public void rxJava07(View view) {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("a");
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                //数据保存到SharedPreferences中
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtil.d(s);
            }
        });

    }

    /**
     * skip表示忽略掉的元素数量。
     *
     * @param view
     */
    public void rxJava08(View view) {
        Observable.just(3, 4, 5, 6, 7)
                .skip(3)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtil.d(integer + "");
                    }
                });
    }

    /**
     * take(count)只取前count个值
     *
     * @param view
     */
    public void rxJava09(View view) {
        Observable.fromArray("a", "b", "c", "d")
                .take(2)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        LogUtil.d(s);
                    }
                });
    }

    /**
     * take(count)只取前count个值
     *
     * @param view
     */
    public void rxJava10(View view) {
        Single.just("a")
                .subscribe(new SingleObserver<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        LogUtil.d("onSubscribe" + d.toString());
                    }

                    @Override
                    public void onSuccess(String s) {
                        LogUtil.d(s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        LogUtil.d("onError" + e.getMessage());
                    }
                });
    }

    /**
     * debounce()去除掉发送太快的事件
     * onNext()事件后的时间>debounce时间，保留该事件
     * 两个相邻数据发射的时间间隔决定了前一个数据是否会被丢弃，最后一个事件后设置的时间无效
     *
     * @param view
     */
    public void rxJava11(View view) {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("a");
                Thread.sleep(700);

                e.onNext("b");
                Thread.sleep(300);

                e.onNext("c");
                Thread.sleep(600);

                e.onNext("d");
                Thread.sleep(500);

                e.onNext("e");
                Thread.sleep(300);
                e.onComplete();
            }
        }).debounce(500, TimeUnit.MILLISECONDS)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        LogUtil.d(s);
                    }
                });
    }

    /**
     * defer:延迟。调用defer()方法后，如果最参数做了修改，那么参数是修改后的值,接收的也是修改后的值。
     * defer:直到观察者Observer订阅它，才会生成新的被观察者Observable。所以订阅前，修改是有效的。订阅的时候才会获取参数的值。
     * 表面是Observable订阅Observer，是为了流式API风格保持不变
     *
     * @param view
     */
    String a = "a";
    String b = "b";
    String c = "c";

    public void rxJava12(View view) {

        Observable<String> observable = Observable.defer(new Callable<ObservableSource<String>>() {
            @Override
            public ObservableSource<String> call() throws Exception {
                LogUtil.d(a + b + c);
                return Observable.just(a, b, c);
            }
        });

        a = "aa";
        b = "bb";
        c = "cc";
        observable.subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(String s) {
                LogUtil.d("onNext:" + s);
            }

            @Override
            public void onError(Throwable e) {
                LogUtil.d("onError:" + e.getMessage());
            }

            @Override
            public void onComplete() {
                LogUtil.d("onComplete");
            }
        });
    }

    /**
     * 不使用defer
     * 接收到的是修改前的值。即使 a  b  c
     *
     * @param view
     */
    public void noDefer(View view) {
        Observable<String> observable = Observable.just(a, b, c);
        a = "aa";
        b = "bb";
        c = "cc";
        observable.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                LogUtil.d("accept:" + s);
            }
        });
    }


    /**
     * last:仅发射最后一个事件，或者满足条件的最后一个事件
     * 参数什么意思？
     *
     * @param view
     */
    public void rxJava13(View view) {
        Observable.just("a", "b", "c")
                .last("a")
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        LogUtil.d("accept:" + s);
                    }
                });
    }

    /**
     * merge:把多个Observable合并成1个然后发射，多个Observable中的元素混合在一起不保证顺序。
     * 不是A发射，B在发射。
     *
     * @param view
     */
    public void rxJava14(View view) {
        Observable.merge(Observable.just(1, 2, 3), Observable.just(4, 5, 6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtil.d("accept:" + integer);
                    }
                });
    }

    /**
     * reduce:把上游发送的多个值处理后发送
     * 第一次发送2个值并处理，之后每次发送一个值并处理，下游只接受处理后的结果
     *
     * @param view
     */
    public void rxJava15(View view) {
        Observable.just(1, 2, 3)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        LogUtil.d("integer=" + integer + ",integer2=" + integer2);
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                LogUtil.d("accept:" + integer);
            }
        });
    }

    /**
     * 发送多个事件，先处理第一个事件，结果仍是第一个事件，再处理第一个事件的结果和第二个事件结果为a,再处理a和第三个事件 结果是b,在处理b和第四个事件结果是c
     * scan()和reduce()类似,每次只发送一个事件。所以apply打印的结果是
     * 1
     * 1 2
     * 3 3
     * <p>
     * 区别：
     * scan(),每发送一次，下游接收一次。
     * reduce()下游只接收最后一次的结果
     *
     * @param view
     */
    public void rxJava16(View view) {
        Observable.just(1, 2, 3)
                .scan(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        LogUtil.d("integer=" + integer + ",integer2=" + integer2);
                        return integer + integer2;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        LogUtil.d("accept:" + integer);
                    }
                });
    }

    /**
     * window(timespan,时间单位)
     * 我理解window的功能和buffer类似 不过buffer 把数据源分割成list 而window把数据源分割成了observable
     *
     * @param view
     */
    public void rxJava17(View view) {
        Observable.interval(1, TimeUnit.SECONDS)
                .take(15)
                .window(3, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Observable<Long>>() {
                    @Override
                    public void accept(Observable<Long> longObservable) throws Exception {
                        LogUtil.d("---------longObservable-----------");
                        longObservable.subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Consumer<Long>() {
                                    @Override
                                    public void accept(Long aLong) throws Exception {
                                        LogUtil.d("accept:" + aLong);
                                    }
                                });
                    }
                });

    }

    /**
     * window(count)
     * 我理解window的功能和buffer类似 不过buffer 把数据源分割成list 而window把数据源分割成了observable
     *
     * @param view
     */
    public void rxJava18(View view) {
        Observable.interval(1, TimeUnit.SECONDS)
                .take(15)
                .window(3)
                .subscribe(new Consumer<Observable<Long>>() {
                    @Override
                    public void accept(Observable<Long> longObservable) throws Exception {
                        LogUtil.d("---------longObservable-----------");
                        longObservable.subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Consumer<Long>() {
                                    @Override
                                    public void accept(Long aLong) throws Exception {
                                        LogUtil.d("accept:" + aLong);
                                    }
                                });
                    }
                });
    }

    /**
     * concat:把多个Observable组合起来，上一个调用onComplete()后，才会发送第二个事件。
     * 应用场景：从换从/网络请求数据
     * 先定义2个Observable:cacheData netData
     * 先从缓存中获取，
     * 缓存为null了，调用onComplete(),这样才会走netData
     * 缓存不为null，不调用onComplete(),不会走netData
     *
     * @param view
     */
    public void rxJava19(View view) {
        Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("a");
                e.onComplete();
            }
        });
        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("b");
            }
        });

        Observable.concat(observable1, observable2)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        LogUtil.d("accept:" + s);
                    }
                });
    }
}
